package streaming.demo3

/**
  * 自定義receiver接收器接收數據
  */

import java.io.{BufferedReader, InputStream, InputStreamReader}
import java.net.Socket
import java.nio.charset.StandardCharsets

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 自定义receiver接收器接收数据
  */
class MyReceiver(host:String,port:Int) extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2){
  //这个方法如何从socket当中接收数据
  def receive()={
    val socket: Socket = new Socket(host,port)
    //获取scoket当中的输入流
    val stream: InputStream = socket.getInputStream
    //将流转换成字符串
    val bufferedReader: BufferedReader = new BufferedReader(new InputStreamReader(stream,StandardCharsets.UTF_8))
  //从bufferedReader里面读取数据
    var line:String=null;
    while ((line=bufferedReader.readLine()) !=null){
      //将我们读取的数据存起来
      store(line)
      line= bufferedReader.readLine()
    }


  }
   override def onStart(): Unit = {
//一旦有数据过来了，我们就会启动一个线程，去接收数据
    new Thread("myReceiver"){
      //线程里面主要用于接收数据
      //使用receiver方法专门从socket当中接收数据
    receive()
    }.start()
  }
//没用
  override def onStop(): Unit = {

  }
}
object MyOwnReceiver {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("MyOwnReceiver")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")
    val streamingContext: StreamingContext = new StreamingContext(sc, Seconds(5))
    //通过自定义receiver接收数据
  val receiverStream: ReceiverInputDStream[String] = streamingContext.receiverStream(new MyReceiver("node01",9999))
    val flatMap: DStream[String] = receiverStream.flatMap(x=>x.split(" "))
    flatMap.print()
    //启动spark程序
    streamingContext.start()
    streamingContext.awaitTermination()
  }
}
